package net.bwie.realtime.dws.log.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.dws.log.bean.AreaControl;
import net.bwie.realtime.dws.log.function.TrafficCarMapFunction;
import net.bwie.realtime.dws.log.function.TrafficViewReportReduceFunction;
import net.bwie.realtime.dws.log.function.TrafficViewReportWindowFunction;
import net.bwie.realtime.dws.log.utils.AreaInfoUtil;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import utils.DorisUtil;
import utils.KafkaUtil;


public class VehicleDistribution {

    public static void main(String[] args) throws Exception{

        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);

        //2.读取数据源
        DataStream<String> trafficStream = KafkaUtil.consumerKafka(env, "traffic_monitor_data");
//        trafficStream.print();

        //3.数据处理
        DataStream<String> resultStream = handle(trafficStream);
        resultStream.print();


        //4.数据输出
        DorisUtil.saveToDoris(resultStream, "car_realtime_report", "dws_vehicle_car_distribution_report");


        //5. 执行
        env.execute("VehicleDistribution");
    }

    private static DataStream<String> handle(DataStream<String> stream) {

        //1.根据区域ID和车辆进行分组
        KeyedStream<String, String> keyStream = stream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String value) throws Exception {
                return JSON.parseObject(value).getString("areaId") +
                        JSON.parseObject(value).getString("car");
            }
        });

        //2.将流中每条数据封装成bean对象
        SingleOutputStreamOperator<AreaControl> mapStream = keyStream.map(new TrafficCarMapFunction());


        //3.分组字段
        KeyedStream<AreaControl, String> keyedStream = mapStream.keyBy(bean -> bean.getAreaId());

        //4.窗口计算
        WindowedStream<AreaControl, String, TimeWindow> windowStream = keyedStream.window(
                TumblingProcessingTimeWindows.of(Time.minutes(1)));


        //5.聚合计算
        SingleOutputStreamOperator<String> applyStream = windowStream.apply(
                new TrafficViewReportReduceFunction(),
                new TrafficViewReportWindowFunction()
        );

        // 3. 调用工具类预加载区域信息
//      AreaInfoUtil.preloadAreaInfo();

        //4. 处理JSON数据（调用工具类的方法）
//       DataStream<String> trafficStream = AreaInfoUtil.processJson(applyStream);

        return applyStream;
    }
}
